package com.bingxu.flink.watermark;

import com.bingxu.flink.bean.WaterSensor;
import com.bingxu.flink.utils.TransFormat;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;

/**
 * @author :BingXu
 * @description :TODO 迟到数据的处理
 * @date :2021/8/16 19:10
 * @modifier :
 */

public class Side_OutStreamSplit {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        env.getConfig().setAutoWatermarkInterval(300);

        OutputTag<String> great50 = new OutputTag<String>("Great50"){};
        OutputTag<WaterSensor> great30 = new OutputTag<WaterSensor>("Great30"){};

        // TODO 每隔5s计算窗口内的WordCount
        SingleOutputStreamOperator<String> processStream = env
                .socketTextStream("localhost", 9999)
                .map((MapFunction<String, WaterSensor>) value -> {
                    String[] fields = value.split(",");
                    return new WaterSensor(fields[0], Long.parseLong(fields[1]) * 1000, Integer.valueOf(fields[2]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((ele, outStamp) -> ele.getTs()))
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor value,
                                               Context ctx,
                                               Collector<String> out) throws Exception {
                        if (value.getVc() > 50) {
                            ctx.output(great50,value.toString());
                        } else if (value.getVc() > 30) {
                            ctx.output(great30, value);
                        } else {
                            out.collect(value.toString());
                        }
                    }
                });

        processStream.print("main");
        processStream.getSideOutput(great50).print("side output 50 ");
        processStream.getSideOutput(great30).print("side output 30");


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
